package com.zxc.kafka.steam;

import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;

@Configuration
public class KafkaStream {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        KStream<String, String> stream = streamsBuilder.stream("stream-topic-input");
        stream.flatMapValues((ValueMapper<String, Iterable<?>>) value-> Arrays.asList(value.split(" ")))
                .groupBy((key, value) -> value).windowedBy(TimeWindows.of(Duration.ofSeconds(1))).count().toStream()
                .map((key, value) -> new KeyValue<>(key.key().toString(), value.toString())).to("stream-topic-out");
        return stream;
    }

//    @Bean
//    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
//        //创建kstream对象，同时指定从那个topic中接收消息
//        KStream<String, String> stream = streamsBuilder.stream("stream-topic-input");
//        stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
//                    @Override
//                    public Iterable<String> apply(String value) {
//                        return Arrays.asList(value.split(" "));
//                    }
//                })
//                //根据value进行聚合分组
//                .groupBy((key,value)->value)
//                //聚合计算时间间隔
//                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
//                //求单词的个数
//                .count()
//                .toStream()
//                //处理后的结果转换为string字符串
//                .map((key,value)->{
//                    System.out.println("key:"+key+",value:"+value);
//                    return new KeyValue<>(key.key().toString(),value.toString());
//                })
//                //发送消息
//                .to("stream-topic-out");
//        return stream;
//    }

}
